package com.yeswater.rabbitmq;

import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import com.yeswater.rabbitmq.utils.RabbitmqUtil;
import com.yeswater.rabbitmq.utils.SleepUtil;

import java.util.concurrent.Callable;
import java.util.concurrent.FutureTask;

/**
 * 工作线程 相当于消费者
 * Author : yeswater
 */

public class Work1 implements Callable {

    public static final String QUEUE_NAME = "mq03";

    public static void main(String[] args) {
        FutureTask futureTask = new FutureTask<>(new Work2());
        Thread thread = new Thread(futureTask);
        thread.start();
    }

    @Override
    public Object call() throws Exception {
        Channel channel = RabbitmqUtil.getChannel();


        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            // 1. 消息标记 tag 2. 批量处理 false
            System.out.println("接收到消息: " + new String(delivery.getBody()));
            SleepUtil.sleep(1);
            System.out.println("处理完消息: " + new String(delivery.getBody()));
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
        };

        CancelCallback cancelCallback = (consumerTag) -> {
            System.out.println(consumerTag + " 消息消费被中断");
        };

        /**
         * 1.消费的队列
         * 2.消费成功之后是否要自动应答
         * 3.消费者未成功消费的回调
         * 4.消费者取消消费的回调
         */
        System.out.println("线程等待接收消息");
        channel.basicConsume(QUEUE_NAME, false, deliverCallback, cancelCallback);
        return null;
    }
}
